package com.microframework.netty;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.protobuf.ProtobufDecoder;
import io.netty.handler.codec.protobuf.ProtobufEncoder;
import io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder;
import io.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.microframework.netty.config.ClientConfig;
import com.microframework.netty.enums.ClientState;
import com.microframework.netty.enums.CommandMessage;
import com.microframework.netty.enums.CommandType;
import com.microframework.netty.manager.SessionManager;
import com.microframework.netty.model.ClientInfo;
import com.microframework.netty.model.ClientSession;

import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
public class EnhancedNettyClient {
    
    private final String host;
    private final int port;
    private final String clientId;
    private final AtomicInteger reconnectAttempts = new AtomicInteger(0);
    private final int maxReconnectAttempts = 10;
    private final long maxReconnectInterval = 60000; // 60 seconds
    
    private Channel channel;
    private EventLoopGroup workerGroup;
    private volatile boolean running = false;
    private volatile ClientState currentState = ClientState.STOPPED;
    
    // 状态监听器列表
    private final List<ClientStateListener> stateListeners = new CopyOnWriteArrayList<>();
    
    // 配置参数
    private final ClientConfig config;
    
    public EnhancedNettyClient(String host, int port) {
        this(host, port, new ClientConfig());
    }
    
    public EnhancedNettyClient(String host, int port, ClientConfig config) {
        this.host = host;
        this.port = port;
        this.clientId = config.getClientId() != null ? config.getClientId() : UUID.randomUUID().toString();
        this.config = config;
    }
    
    public synchronized void start() {
        if (running) {
            log.warn("Client is already running");
            return;
        }
        
        running = true;
        reconnectAttempts.set(0);
        workerGroup = new NioEventLoopGroup(config.getWorkerThreads());
        
        try {
            Bootstrap b = new Bootstrap();
            b.group(workerGroup)
             .channel(NioSocketChannel.class)
             .option(ChannelOption.SO_KEEPALIVE, true)
             .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, config.getConnectTimeout())
             .option(ChannelOption.TCP_NODELAY, true)
             .handler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 protected void initChannel(SocketChannel ch) {
                     ChannelPipeline pipeline = ch.pipeline();
                     
                     // SSL/TLS 支持
                     if (config.isSslEnabled()) {
                         pipeline.addLast(config.getSslContext().newHandler(ch.alloc(), host, port));
                     }
                     
                     // 空闲检测
                     pipeline.addLast(new IdleStateHandler(
                         config.getReaderIdleTime(), 
                         config.getWriterIdleTime(), 
                         config.getAllIdleTime(), 
                         TimeUnit.SECONDS));
                     pipeline.addLast(new HeartbeatHandler(EnhancedNettyClient.this));
                     
                     // 编解码
                     pipeline.addLast(new LengthFieldPrepender(4));
                     pipeline.addLast(new LengthFieldBasedFrameDecoder(
                         config.getMaxFrameLength(), 
                         0, 4, 0, 4));
                  // 添加 Protobuf 编码解码器
                     pipeline.addLast(new ProtobufVarint32FrameDecoder());
                     pipeline.addLast(new ProtobufDecoder(YourMessageType.getDefaultInstance()));
                     pipeline.addLast(new ProtobufVarint32LengthFieldPrepender());
                     pipeline.addLast(new ProtobufEncoder());
                     
                     // 业务处理器
                     pipeline.addLast(new ClientCommandHandler(EnhancedNettyClient.this));
                     pipeline.addLast(new ClientStateHandler(EnhancedNettyClient.this));
                 }
             });
            
            // 连接服务器
            connect(b);
            
        } catch (Exception e) {
            log.error("Client start failed", e);
            workerGroup.shutdownGracefully();
            running = false;
            setState(ClientState.STOPPED);
        }
    }
    
    private void connect(Bootstrap bootstrap) {
        if (!running) return;
        
        log.info("Connecting to server {}:{}", host, port);
        setState(ClientState.CONNECTING);
        
        bootstrap.connect(host, port).addListener((ChannelFuture future) -> {
            if (future.isSuccess()) {
                channel = future.channel();
                reconnectAttempts.set(0);
                log.info("Connected to server successfully");
                setState(ClientState.CONNECTED);
                
                // 注册会话
                ClientSession session = new ClientSession(channel, clientId);
                SessionManager.addSession(session);
                
                // 发送客户端信息
                sendClientInfo();
                
                // 启动状态报告定时任务
                if (config.isStatusReportEnabled()) {
                    scheduleStatusReport();
                }
                
                // 监听连接关闭
                channel.closeFuture().addListener(closeFuture -> {
                    log.info("Connection closed");
                    setState(ClientState.DISCONNECTED);
                    SessionManager.removeSession(channel);
                    scheduleReconnect();
                });
                
            } else {
                log.error("Connection failed: {}", future.cause().getMessage());
                setState(ClientState.CONNECT_FAILED);
                scheduleReconnect();
            }
        });
    }
    
    private void scheduleReconnect() {
        if (!running) return;
        
        int attempts = reconnectAttempts.incrementAndGet();
        if (attempts > maxReconnectAttempts) {
            log.error("Max reconnect attempts ({}) reached, stopping client", maxReconnectAttempts);
            stop();
            return;
        }
        
        // 指数退避算法
        long delay = Math.min((long) (Math.pow(2, attempts) * 1000), maxReconnectInterval);
        
        log.info("Attempting to reconnect (attempt {}/{}), waiting {} ms...", 
                attempts, maxReconnectAttempts, delay);
        
        setState(ClientState.RECONNECTING);
        
        workerGroup.schedule(() -> {
            if (running) {
                connect(new Bootstrap().group(workerGroup)
                    .channel(NioSocketChannel.class)
                    .handler(channel.pipeline().last()));
            }
        }, delay, TimeUnit.MILLISECONDS);
    }
    
    private void scheduleStatusReport() {
        if (channel != null && channel.isActive()) {
            channel.eventLoop().scheduleAtFixedRate(() -> {
                if (isConnected()) {
                    Map<String, Object> status = new StatusCollector().collect();
                    sendCommand(CommandType.CLIENT_STATUS, status);
                }
            }, config.getStatusReportInitialDelay(), 
               config.getStatusReportInterval(), 
               TimeUnit.SECONDS);
        }
    }
    
    private void sendClientInfo() {
        if (channel != null && channel.isActive()) {
            ClientInfo clientInfo = new ClientInfo(
                clientId, 
                config.getClientVersion(), 
                System.currentTimeMillis(),
                config.getClientType(),
                config.getMetadata()
            );
            sendCommand(CommandType.CLIENT_INFO, clientInfo);
        }
    }
    
    public boolean sendCommand(CommandType type, Object payload) {
        if (channel != null && channel.isActive()) {
            CommandMessage command = new CommandMessage(type, payload);
            channel.writeAndFlush(command).addListener(future -> {
                if (!future.isSuccess()) {
                    log.error("Failed to send command: {}", future.cause().getMessage());
                }
            });
            return true;
        } else {
            log.warn("Cannot send command, channel is not active");
            return false;
        }
    }
    
    public synchronized void stop() {
        if (!running) return;
        
        log.info("Stopping client...");
        running = false;
        
        if (channel != null) {
            channel.close().addListener(future -> {
                if (!future.isSuccess()) {
                    log.error("Error closing channel: {}", future.cause().getMessage());
                }
            });
        }
        
        if (workerGroup != null) {
            workerGroup.shutdownGracefully(
                config.getShutdownQuietPeriod(), 
                config.getShutdownTimeout(), 
                TimeUnit.MILLISECONDS)
                .addListener(future -> {
                    if (!future.isSuccess()) {
                        log.error("Error shutting down worker group: {}", future.cause().getMessage());
                    }
                });
        }
        
        setState(ClientState.STOPPED);
    }
    
    public void addStateListener(ClientStateListener listener) {
        stateListeners.add(listener);
    }
    
    public void removeStateListener(ClientStateListener listener) {
        stateListeners.remove(listener);
    }
    
    private void setState(ClientState newState) {
        ClientState oldState = this.currentState;
        this.currentState = newState;
        
        if (oldState != newState) {
            log.info("Client state changed from {} to {}", oldState, newState);
            notifyStateChanged(newState);
        }
    }
    
    private void notifyStateChanged(ClientState state) {
        stateListeners.forEach(listener -> {
            try {
                listener.onStateChanged(state);
            } catch (Exception e) {
                log.error("Error notifying state listener", e);
            }
        });
    }
    
    private void notifyCommandReceived(CommandMessage command) {
        stateListeners.forEach(listener -> {
            try {
                listener.onCommandReceived(command);
            } catch (Exception e) {
                log.error("Error notifying command listener", e);
            }
        });
    }
    
    public boolean isConnected() {
        return currentState == ClientState.CONNECTED && 
               channel != null && 
               channel.isActive();
    }
    
    public ClientState getCurrentState() {
        return currentState;
    }
    
    public String getClientId() {
        return clientId;
    }
    
    
    // 状态监听器接口
    public interface ClientStateListener {
        void onStateChanged(ClientState state);
        void onCommandReceived(CommandMessage command);
    }
    
    // 内部处理器类
    private static class ClientStateHandler extends ChannelInboundHandlerAdapter {
        private final EnhancedNettyClient client;
        
        public ClientStateHandler(EnhancedNettyClient client) {
            this.client = client;
        }
        
        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            client.setState(ClientState.DISCONNECTED);
            super.channelInactive(ctx);
        }
        
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            log.error("Channel error: {}", cause.getMessage());
            ctx.close();
        }
    }
    
    private static class ClientCommandHandler extends ChannelInboundHandlerAdapter {
        private final EnhancedNettyClient client;
        
        public ClientCommandHandler(EnhancedNettyClient client) {
            this.client = client;
        }
        
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            if (msg instanceof CommandMessage) {
                CommandMessage command = (CommandMessage) msg;
                client.notifyCommandReceived(command);
                handleCommand(command, ctx);
            } else {
                log.warn("Received unknown message type: {}", msg.getClass().getName());
                // 可以选择回复错误消息
                CommandMessage errorResponse = new CommandMessage(
                    CommandType.ERROR, 
                    "Unsupported message type: " + msg.getClass().getName());
                ctx.writeAndFlush(errorResponse);
            }
        }
        
        private void handleCommand(CommandMessage command, ChannelHandlerContext ctx) {
            try {
                switch (command.getType()) {
                    case SERVER_TIME_SYNC:
                        handleTimeSync(command.getPayload());
                        break;
                    case SERVER_CONFIG_UPDATE:
                        handleConfigUpdate(command.getPayload());
                        break;
                    case SERVER_RESTART_COMMAND:
                        handleRestartCommand(command.getPayload(), ctx);
                        break;
                    case SERVER_PING:
                        handlePing(command, ctx);
                        break;
                    case ERROR:
                        handleError(command.getPayload());
                        break;
                    default:
                        log.warn("Unknown command type: {}", command.getType());
                        // 回复未知命令错误
                        CommandMessage errorResponse = new CommandMessage(
                            CommandType.ERROR, 
                            "Unknown command type: " + command.getType());
                        ctx.writeAndFlush(errorResponse);
                }
            } catch (Exception e) {
                log.error("Error handling command: {}", e.getMessage());
                CommandMessage errorResponse = new CommandMessage(
                    CommandType.ERROR, 
                    "Error processing command: " + e.getMessage());
                ctx.writeAndFlush(errorResponse);
            }
        }
        
        private void handleTimeSync(Object payload) {
            if (payload instanceof Long) {
                long serverTime = (Long) payload;
                log.info("Server time sync: {}", new Date(serverTime));
                // 可以在这里实现客户端时间同步逻辑
            } else {
                log.warn("Invalid time sync payload: {}", payload);
            }
        }
        
        private void handleConfigUpdate(Object payload) {
            if (payload instanceof Map) {
                @SuppressWarnings("unchecked")
                Map<String, Object> config = (Map<String, Object>) payload;
                log.info("Received config update: {}", config);
                // 更新客户端配置
            } else {
                log.warn("Invalid config update payload: {}", payload);
            }
        }
        
        private void handleRestartCommand(Object payload, ChannelHandlerContext ctx) {
            log.info("Received restart command");
            // 先确认收到命令
            ctx.writeAndFlush(new CommandMessage(CommandType.ACK, "Restart command received"));
            
            // 延迟执行重启
            ctx.channel().eventLoop().schedule(() -> {
                System.exit(0);
            }, 1, TimeUnit.SECONDS);
        }
        
        private void handlePing(CommandMessage command, ChannelHandlerContext ctx) {
            log.debug("Received ping, sending pong");
            ctx.writeAndFlush(new CommandMessage(CommandType.PONG, command.getTimestamp()));
        }
        
        private void handleError(Object payload) {
            log.error("Received error from server: {}", payload);
        }
    }
    
    private static class HeartbeatHandler extends ChannelInboundHandlerAdapter {
        private static final CommandMessage HEARTBEAT_MSG = 
            new CommandMessage(CommandType.HEARTBEAT, null);
        
        private final EnhancedNettyClient client;
        
        public HeartbeatHandler(EnhancedNettyClient client) {
            this.client = client;
        }
        
        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            if (evt instanceof IdleStateEvent) {
                IdleStateEvent e = (IdleStateEvent) evt;
                if (e.state() == IdleState.WRITER_IDLE) {
                    if (client.isConnected()) {
                        ctx.writeAndFlush(HEARTBEAT_MSG).addListener(future -> {
                            if (!future.isSuccess()) {
                                log.error("Heartbeat send failed", future.cause());
                            }
                        });
                    }
                } else if (e.state() == IdleState.READER_IDLE) {
                    log.warn("No message received from server for {} seconds, closing connection", 
                            client.config.getReaderIdleTime());
                    ctx.close();
                }
            } else {
                super.userEventTriggered(ctx, evt);
            }
        }
    }
    
    
    // 状态收集器
    private static class StatusCollector {
        public Map<String, Object> collect() {
            // 实现状态收集逻辑
            return Map.of(
                "timestamp", System.currentTimeMillis(),
                "status", "normal",
                "resources", Map.of(
                    "cpu", Runtime.getRuntime().availableProcessors(),
                    "memory", Runtime.getRuntime().freeMemory()
                )
            );
        }
    }
}